package com.bungarus.busi;

import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.config.BusiSpringConfig;
import com.bungarus.helper.RedisCacheHelper;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageType;
import com.bungarus.model.Session;
import io.lettuce.core.KeyValue;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import io.netty.channel.ChannelHandlerContext;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.util.StringUtils;

import java.util.*;

/**
 * Session management for authenticated user. It is a shared memory db supported by Redis that stores
 * all authenticated user session.
 * All methods in this type are asynchronous to ensure the large throughout of the server.
 *
 * The patter of session cached in redis likes:
 *   set  -> user:tenantid:userid dev1+app1 dev2+app2 dev3+app3
 *   hash -> tenantid:userid:dev1+app1 sessionid xxx ip_port xxx
 *
 * Created by tang on 2018/12/16.
 */
public class SessionManagement extends BusiProcessing {
    Logger logger = Logger.getLogger(SessionManagement.class);

    private static final String SESSION_TAG = "session_id";
    private static final String SESSION_IPPORT = "ip_port";
    private static final long SESSION_EXPIRE = 2 * 3600 * 1000;

    private static final String PREFIX_USER = "user:";
    private static final String APP_ALL = "ALL";
    private static final String DEV_ALL = "ALL";

    static final String BUSI_SERVER = "busiServer";
    static final String VALUE_SEPERATOR = ",";

    public final static ApplicationContext springCtx = new AnnotationConfigApplicationContext(BusiSpringConfig.class);

    private RedisClient sessionCache;
    private StatefulRedisConnection<String, String> connection;
    private StatefulRedisConnection<String, String> connection4Multi;

    private static SessionManagement sessionManagement = new SessionManagement();
    private SessionManagement(){
        init();
    }

    private void init() {
        sessionCache = RedisCacheHelper.initRedisClient();
        connection = sessionCache.connect();
        connection4Multi = sessionCache.connect();

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                RedisCacheHelper.shutdownRedisClient(sessionCache);
            }
        });
    }

    public static SessionManagement getInstance() {
        return sessionManagement;
    }

    private StatefulRedisConnection<String, String> getConnection() {
        return connection;
    }

    private StatefulRedisConnection<String, String> getConnection4Multi() {
        return connection4Multi;
    }

    @Override
    public void doBusiness(Object o) throws BusiProcessingException {
        if(null == o || !(o instanceof Map)) {
            throw new IllegalArgumentException("Input argument must be instance of Map.");
        }
        Map map = (Map)o;
        ChannelHandlerContext ctx = (ChannelHandlerContext)map.get(SESSION_CONTEXT);
        MessageProtobuf.Message sessionMsg = (MessageProtobuf.Message)map.get(SESSION);

        if(sessionMsg.getHeader().getType() == MessageType.AUTH) {
            for (MessageProtobuf.Extension extension : sessionMsg.getHeader().getExtensionsList()) {
                if (extension.getKey().equalsIgnoreCase(Session.CLIENT)) {
                    String ip_port = extension.getValue();
                    save(
                            sessionMsg.getHeader().getFrom(),
                            String.valueOf(sessionMsg.getHeader().getTenantID()),
                            sessionMsg.getHeader().getApp(),
                            sessionMsg.getHeader().getDev(),
                            //user id which has prefix "user:"
                            sessionMsg.getHeader().getSessionId(),  // session id
                            ip_port);   // ip and port of access node to which the message will be sent

                    //Send ack signal to transfer node when session info has been saved successfully.
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(sessionMsg.getHeader().getId(), sessionMsg.getHeader().getId2()));
                    break;
                }
            }
        }
        else if(sessionMsg.getHeader().getType() == MessageType.LOGOUT) {
            for (MessageProtobuf.Extension extension : sessionMsg.getHeader().getExtensionsList()) {
                if (extension.getKey().equalsIgnoreCase(Session.CLIENT)) {
                    String ip_port = extension.getValue();
                    remove(
                            sessionMsg.getHeader().getFrom(),
                            String.valueOf(sessionMsg.getHeader().getTenantID()),
                            sessionMsg.getHeader().getApp(),
                            sessionMsg.getHeader().getDev(),
                            //user id which has prefix "user:"
                            sessionMsg.getHeader().getSessionId(),  // session id
                            ip_port);   // ip and port of access node to which the message will be sent

                    //Send ack signal to transfer node when session info has been removed successfully.
                    ctx.writeAndFlush(MessageBuilder.buildAckMessage(sessionMsg.getHeader().getId(), sessionMsg.getHeader().getId2()));
                    break;
                }
            }
        }
    }

    /**
     * save session data
     * @param userId
     * @param tenantId
     * @param app
     * @param dev
     * @param sessionId
     * @param ipport
     */
    public void save(String userId, String tenantId, String app, String dev, String sessionId, String ipport) {
        //cache key pattern: user:{tenantId}:{userId}
        //cache value pattern: {userId},{sessionId},{ipport}
        //ipport format: {ip}:{port}
        String key = getSessionKey(userId, tenantId);
        String hashKey = getSessionHashKey(userId, tenantId, app, dev);

        StatefulRedisConnection<String, String> connection = getConnection4Multi();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        asyncCommands.setAutoFlushCommands(false); //write commands into buffer first

        //using redis's transaction function to fulfil the atomic operation
        asyncCommands.multi();
        asyncCommands.sadd(key, getAppStr(app) + ":" + getDevStr(dev));
        asyncCommands.hset(hashKey, SESSION_TAG, sessionId);
        asyncCommands.hset(hashKey, SESSION_IPPORT, ipport);
        asyncCommands.expire(key, SESSION_EXPIRE / 1000);
        asyncCommands.expire(hashKey, SESSION_EXPIRE / 1000);
        asyncCommands.exec();

        asyncCommands.flushCommands();
        logger.debug("saved session:" + sessionId);
    }

    /**
     * get session data
     * @param userId
     * @param tenantId
     * @return
     */
    public Set<Object> retrieve(String userId, String tenantId) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        RedisFuture<Set<String>> devAppSetFuture = asyncCommands.smembers(getSessionKey(userId, tenantId));
        connection.flushCommands();

        Set<String> devAppSet = RedisCacheHelper.getFutureValue(devAppSetFuture);
        Set<Object> result = new LinkedHashSet<>();
        if(null != devAppSet) {
            devAppSet.stream().forEach(devApp -> {
                //pattern: {dev}:{app}
                String hashKey = tenantId + ":" + userId + ":" + devApp;

                RedisFuture<List<KeyValue<String, String>>> mediumFuture = asyncCommands.hmget(hashKey, SESSION_TAG, SESSION_IPPORT);
                connection.flushCommands();

                List<KeyValue<String, String>> medium = RedisCacheHelper.getFutureValue(mediumFuture);
                if (null != medium && medium.size() > 0) {
                    //pattern: {userId},{sessionId},{ipport}
                    result.add(userId + VALUE_SEPERATOR + medium.get(0).getValue() + VALUE_SEPERATOR + medium.get(1).getValue());
                }
            });
        }
        return result;
    }

    public List<Object> retrieveM(List<String> userIds, String tenantId) {
        logger.debug("retrieve multiple sessions");
        List<Object> results = new ArrayList<>();
        userIds.stream().forEach(userId -> {
            Set result = retrieve(userId, tenantId);
            if(result.size() > 0) {
                results.add(result);
            }
        });
        return results;
    }

    public Optional<String> retrieve(String userId, String tenantId, String app, String dev) {
        StatefulRedisConnection<String, String> connection = getConnection();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();

        String hashKey = getSessionHashKey(userId, tenantId, app, dev);

        RedisFuture<List<KeyValue<String, String>>> mediumFuture = asyncCommands.hmget(hashKey, SESSION_TAG, SESSION_IPPORT);
        connection.flushCommands();

        List<KeyValue<String, String>> medium = RedisCacheHelper.getFutureValue(mediumFuture);

        String result = null;
        if(null != medium && medium.size() > 0) {
            //pattern: {userId},{sessionId},{ipport}
            result = userId + VALUE_SEPERATOR + medium.get(0).getValue() + VALUE_SEPERATOR + medium.get(1).getValue();
        }

        return Optional.ofNullable(result);
    }


    /**
     * remove session data
     * @param userId
     * @param tenantId
     * @param app
     * @param dev
     * @param sessionId
     * @param ipport
     */
    public void remove(String userId, String tenantId, String app, String dev, String sessionId, String ipport) {
        logger.debug("remove session:" + sessionId);
        //cache key pattern: user:{tenantId}:{userId}
        //cache value pattern: {userId},{sessionId},{ipport}
        //ipport format: {ip}:{port}
        String key = getSessionKey(userId, tenantId);
        String hashKey = getSessionHashKey(userId, tenantId, app, dev);
        StatefulRedisConnection<String, String> connection = getConnection4Multi();
        RedisAsyncCommands<String, String> asyncCommands = connection.async();
        asyncCommands.setAutoFlushCommands(false);

        //using redis's transaction function to fulfil the atomic operation
        asyncCommands.multi();
        asyncCommands.srem(key, getAppStr(app) + ":" + getDevStr(dev));
        asyncCommands.hdel(hashKey, SESSION_TAG, SESSION_IPPORT);
        asyncCommands.exec();

        asyncCommands.flushCommands();
    }

    private String getSessionKey(String userId, String tenantId){
        return PREFIX_USER + tenantId + ":" + userId;
    }

    private String getSessionHashKey(String userId, String tenantId, String app, String dev){
        return tenantId + ":" + userId + ":" + getAppStr(app) + ":" + getDevStr(dev);
    }

    private String  getAppStr(String app){
        return StringUtils.isEmpty(app) ? APP_ALL : app;
    }

    private String getDevStr(String dev){
        return StringUtils.isEmpty(dev) ? DEV_ALL : dev;
    }
}
